package com.echo.code.thread.mode.listener;

import java.lang.reflect.Method;
import java.util.Objects;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;

/**
 * Dispatcher Created by baiyap on 2018/9/24.
 */
public class Dispatcher {

    private final ExecutorService executorService;
    private final EventExceptionHandler exceptionHandler;

    public Dispatcher(ExecutorService executorService, EventExceptionHandler exceptionHandler) {
        this.executorService = executorService;
        this.exceptionHandler = exceptionHandler;
    }

    public void dispatch(Registry registry, Object event, String topic) {
        // 根据topic获取所有的Subscriber列表
        ConcurrentLinkedQueue<Subscriber> subscribers = registry.scanSubscriber(topic);
        if (Objects.isNull(subscribers)) {
            if (Objects.nonNull(exceptionHandler)) {
                exceptionHandler.handle(new IllegalArgumentException("The topic " + topic + " not bind yet"));
            }
            return;
        }

        // 遍历所有的方法, 并且通过反射的方式进行方法调用
        subscribers.stream().filter(subscriber -> !subscriber.isDisable()).filter(subscriber -> {
            Method subscribeMethod = subscriber.getSubscribeMethod();
            Class<?> clazz = subscribeMethod.getParameterTypes()[0];
            return (clazz.isAssignableFrom(event.getClass()));
        }).forEach(subscriber -> realInvokeSubscribe(subscriber, event));
    }

    private void realInvokeSubscribe(Subscriber subscriber, Object event) {
        Method subscribeMethod = subscriber.getSubscribeMethod();
        Object subscribeObject = subscriber.getSubscribeObject();
        executorService.execute(() -> {
            try {
                subscribeMethod.invoke(subscribeObject, event);
            } catch (Exception e) {
                if (Objects.nonNull(exceptionHandler)) {
                    exceptionHandler.handle(e);
                }
            }
        });
    }

    public void close() {
        executorService.shutdown();
    }
}
